/*
 * Copyright 2013 JROSE
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.jrose.core.jdbc.persistence.support;

import lombok.AccessLevel;
import lombok.Getter;
import lombok.NoArgsConstructor;
import org.jrose.api.annotation.cache.CacheKey;
import org.jrose.api.annotation.cache.CacheSupport;
import org.jrose.api.annotation.cache.Cacheable;
import org.jrose.api.common.IterableFilter;
import org.jrose.core.cache.CacheEvictor;
import org.jrose.core.cache.CacheEvitorCallback;
import org.jrose.core.cache.CacheSupportProvider;
import org.jrose.core.cache.EvictCandidateProvider;
import org.jrose.core.cache.generator.CacheNameGenerator;
import org.jrose.core.jdbc.datasource.RoutingPolicy;
import org.jrose.core.jdbc.datasource.UseRoutingPolicy;
import org.jrose.core.jdbc.persistence.query.Query;
import org.jrose.core.util.AnnotationUtils;
import org.jrose.core.util.CollectionUtils;
import org.springframework.cache.Cache;
import org.springframework.util.Assert;

import java.io.Serializable;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Abstract cacheable persistence implementation.
 *
 * @author Xiaohai Zhang
 * @since 2013-08-18 20:38
 */
@NoArgsConstructor(access = AccessLevel.PROTECTED)
abstract public class CacheablePersistence<E extends Serializable> extends SqlExecutorProvider<E> implements CacheSupportProvider {
    @Getter protected CacheSupport cacheSupport;
    @Getter protected boolean cacheable;

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet();

        // initialize @CacheSupport annotation if necessary
        this.cacheSupport = null;
        CacheSupport annotation = AnnotationUtils.getAnnotation(getClass(), CacheSupport.class);
        if (annotation != null && annotation.enable()) {
            this.cacheSupport = annotation;
        }
        this.cacheable = this.cacheSupport != null;
    }

    @Override
    @Cacheable
    @UseRoutingPolicy(RoutingPolicy.usingMaster)
    public E load(@CacheKey Serializable... ids) {
        validatePrimaryKey(ids);
        return super.load(ids);
    }

    @Override
    public E persist(E entity) {
        if (!isCacheable()) {
            return super.persist(entity);
        } else {
            final E result = super.persist(entity);
            evictFromCache(new EvictCandidateProvider<E>() {
                @Override
                public List<E> provide() {
                    return Collections.singletonList(result);
                }
            });
            return result;
        }
    }

    @Override
    public void update(E entity, Serializable... ids) {
        validatePrimaryKey(ids);
        if (!isCacheable()) {
            super.update(entity, ids);
        } else {
            evictFromCache(ids);
            super.update(entity, ids);
            evictFromCache(ids);
        }
    }

    @Override
    public void delete(Serializable... ids) {
        validatePrimaryKey(ids);
        if (!isCacheable()) {
            super.delete(ids);
        } else {
            evictFromCache(ids);
            super.delete(ids);
        }
    }

    abstract protected void generateCacheKeys(E source, List<String> keys);

    protected void evictFromCache(Serializable... ids) {
        if (!isCacheable()) {
            return;
        }
        final E entity = load(ids);
        if (entity == null) {
            return;
        }
        evictFromCache(new EvictCandidateProvider<E>() {
            @Override
            public List<E> provide() {
                return Collections.singletonList(entity);
            }
        });
    }

    protected void evictFromCache(EvictCandidateProvider<E> provider) {
        if (!isCacheable()) {
            return;
        }
        String cacheName = CacheNameGenerator.generate(this);
        if (cacheName == null) {
            return;
        }
        Cache cache = cacheManager.getCache(cacheName);
        if (cache == null) {
            return;
        }
        if (provider == null || provider.provide() == null) {
            return;
        }
        List<String> keys = new LinkedList<>();
        for (E source : provider.provide()) {
            if (source == null) {
                continue;
            }
            generateCacheKeys(source, keys);
        }
        keys = CollectionUtils.filterIterable(keys, new IterableFilter<String>() {
            @Override
            public boolean accept(String element) {
                return element != null;
            }
        });

        for (String key : keys) {
            try {
                cache.evict(key);
            } catch (Throwable ignored) {
                log.warn(ignored.getMessage(), ignored);
            }
        }
    }


    abstract public static class AbstractCacheEvictor implements CacheEvictor {
        private final AtomicBoolean closed = new AtomicBoolean(false);

        @Override
        final public void close() {
            evict();
        }

        @Override
        final public void evict() {
            if (closed.get()) {
                // avoid duplicated invocation on doEvict method
                return;
            }
            try {
                doEvict();
            } finally {
                closed.set(true);
            }
        }

        @Override
        final public <T> T execute(CacheEvitorCallback<T> callback) {
            Assert.notNull(callback);
            try {
                return callback.withinCacheEvitor();
            } finally {
                evict();
            }
        }

        abstract protected void doEvict();
    }

    public CacheEvictor generateCacheEvictor(final Serializable... ids) {
        EvictCandidateProvider<E> provider = new EvictCandidateProvider<E>() {
            @Override
            public List<E> provide() {
                E entity = load(ids);
                return entity == null ? Collections.<E>emptyList() : Collections.singletonList(entity);
            }
        };
        final List<E> candidates = provider.provide();
        return new AbstractCacheEvictor() {
            @Override
            protected void doEvict() {
                if (candidates == null || candidates.isEmpty()) {
                    return;
                }
                evictFromCache(new EvictCandidateProvider<E>() {
                    @Override
                    public List<E> provide() {
                        return candidates;
                    }
                });
            }
        };
    }

    public CacheEvictor generateCacheEvictor(final Query<E> query) {
        EvictCandidateProvider<E> provider = new EvictCandidateProvider<E>() {
            @Override
            public List<E> provide() {
                return executeQuery(query);
            }
        };
        final List<E> candidates = provider.provide();
        return new AbstractCacheEvictor() {
            @Override
            protected void doEvict() {
                if (candidates == null || candidates.isEmpty()) {
                    return;
                }
                evictFromCache(new EvictCandidateProvider<E>() {
                    @Override
                    public List<E> provide() {
                        return candidates;
                    }
                });
            }
        };
    }
}
